package easy.trace.core.support;


import com.alibaba.fastjson.JSON;
import easy.trace.common.Span;
import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author easy
 */
public class RingBuffer {
    private final Logger logger = Logger.getLogger(RingBuffer.class);

    private LinkedBlockingQueue<Span> queues;
    private AtomicBoolean startSend = new AtomicBoolean(false);
    private TransferTask transferTask;
    private List<Span> spanCache;
    private static final int QUEUE_CAPACITY = 100;

    private boolean isReady() {
        return startSend.get();
    }

    public boolean add(Span span) {
        if (isReady() && queues.size() < QUEUE_CAPACITY) {
            return queues.add(span);
        } else {
            return false;
        }
    }

    public RingBuffer() {
        this.queues = new LinkedBlockingQueue<>(QUEUE_CAPACITY);
        this.spanCache = new ArrayList<>();
        this.transferTask = new TransferTask();
    }

    private class TransferTask extends Thread {


        TransferTask() {
            this.setName(TransferTask.class.getName());
        }

        /**
         * 优化使用传统rpc 方式
         * carrier.push(spanCache);
         */
        @Override
        public void run() {
            for (; ; ) {
                if (!isReady()) {
                    continue;
                }
                try {
                    Span first = queues.take();
                    queues.drainTo(spanCache);
                    spanCache.add(first);

                    for (Span span : spanCache) {
                        logger.info(JSON.toJSONString(span));
                    }

                    spanCache.clear();
                } catch (Exception e) {
                    e.printStackTrace();
                    startSend.set(false);
                }
            }
        }
    }

    public void start() {
        if (!transferTask.isAlive()) {
            transferTask.start();
            Runtime.getRuntime().addShutdownHook(new Thread() {
                @Override
                public void run() {
                    cancel();
                }
            });
            startSend.set(true);
        }
    }

    public void cancel() {
        transferTask.interrupt();
    }

}
